﻿using CanalService.Model;
using CanalSharp.Connections;
using CanalSharp.Protocol;
using Microsoft.Extensions.Caching.Distributed;
using System.Reflection;
using System.Text.Json;
using LogLevel = Microsoft.Extensions.Logging.LogLevel;
using Type = System.Type;

namespace CanalService.Services
{
    /// <summary>
    /// 同步MySQL数据到Redis
    /// </summary>
    public class CanalToRedisService
    {

        //依赖注入redis分布式缓存
        private readonly IDistributedCache _distributedCache;
        
        public CanalToRedisService(IDistributedCache distributedCache)
        {
            _distributedCache = distributedCache;
        }

        /// <summary>
        /// 监听服务
        /// </summary>
        public async void OnStart()
        {
            //1、日志准备
            var loggerFactory = LoggerFactory.Create(builder =>
            {
                builder
                   .AddFilter("Microsoft", LogLevel.Debug)
                   .AddFilter("System", LogLevel.Information)
                   .AddConsole();
            });

            var logger = loggerFactory.CreateLogger<SimpleCanalConnection>();
            //1、建立连接,
            var conn = new SimpleCanalConnection(new SimpleCanalOptions("127.0.0.1", 11111, "101"), logger);
            await conn.ConnectAsync();

            //2、订阅数据库中所有库和所有表
            await conn.SubscribeAsync(".*\\..*");

            // 3、读取canal数据
            while (true)
            {

                try
                {
                    //3.1、分段读取到canal数据
                    var messsage = await conn.GetWithoutAckAsync(1024);

                    //4、判断数据类型(加锁和不加锁)
                    var entries = messsage.Entries;
                    foreach (var entry in entries)
                    {
                        //不处理事务标记
                        if (entry.EntryType == EntryType.Transactionbegin || entry.EntryType == EntryType.Transactionend)
                        {
                            continue;
                        }

                        // 5、进一步判断数据操作类型（增删改查）
                        RowChange rowChange = RowChange.Parser.ParseFrom(entry.StoreValue);

                        // 5.1、数据增加状态
                        if (rowChange.EventType == EventType.Insert)
                        {
                            logger.LogInformation("-------> 增加数据");

                            // 6、获取具体字段和数据
                            foreach (var rowData in rowChange.RowDatas)
                            {
                                //6.1、获取所有表数据
                                List<Column> columns = rowData.AfterColumns.ToList();

                                // 6.3、获取数据库
                                string SchemaName = entry.Header.SchemaName;

                                // 6.3、获取数据库对应的表
                                string tableName = entry.Header.TableName;

                                //判断数据库名是ncd_test并且表名是canal_table
                                if (SchemaName.Equals("ndc_test") && tableName.Equals("canal_table"))
                                {
                                    // 6.2、同步数据
                                    SyncCanalToRedis(columns);
                                }
                            }
                        } 

                        if(rowChange.EventType == EventType.Delete)
                        {
                            logger.LogInformation("-----------删除数据-----------");
                        }

                    }

                }
                catch (Exception e)
                {
                    logger.LogError(e, "Error.");
                    //发生异常执行重连，此方法只有集群连接对象才有
                }
            }

        }

        private void SyncCanalToRedis(List<Column> columns)
        {
            try
            {
                //1、创建对象
                CanalTable model = new();

                //2、数据转换
                Type type = typeof(CanalTable);
                foreach (var column in columns)
                {

                    string filedName = column.Name;
                    string filedValue = column.Value;

                    //3、反射读取具体属性
                    PropertyInfo propertyInfo = type.GetProperty(filedName);
                    //4、数据类型确认
                    if (propertyInfo != null)
                    {
                        Type type1 = propertyInfo.PropertyType;
                        if (type1 == typeof(int))
                        {
                            propertyInfo.SetValue(model, int.Parse(filedValue));
                        }
                        else if (type1 == typeof(string))
                        {
                            propertyInfo.SetValue(model, filedValue);
                        }
                        else if (type1 == typeof(double))
                        {
                            propertyInfo.SetValue(model, double.Parse(filedValue));
                        }
                        else if (type1 == typeof(bool))
                        {
                            propertyInfo.SetValue(model, bool.Parse(filedValue));
                        }
                        else if (type1 == typeof(DateTime))
                        {
                            propertyInfo.SetValue(model, DateTime.Parse(filedValue));
                        }
                        else if (type1 == typeof(long))
                        {
                            propertyInfo.SetValue(model, long.Parse(filedValue));
                        }
                        else
                        {
                            Console.WriteLine("不知道是什么类型");
                        }
                    }
                }

                string productString = JsonSerializer.Serialize(model);
                var cacheKey = Guid.NewGuid().ToString("N");
                _distributedCache.SetString(cacheKey, productString);

            }
            catch (Exception)
            {

                throw;
            }

        }

    }
}
